/**
 * 
 */
package SRTPlanner;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import simulation.ConfigGraph;

/**
 * @author aclayton
 *
 */
public class ParrallelizedSRT {
	public static class ParrallelizedSRTMapper extends MapReduceBase
	 implements Mapper<IntWritable, ConfigGraph, IntWritable, Text>{

		@Override
		public void map(IntWritable arg0, ConfigGraph arg1,
				OutputCollector<IntWritable, Text> arg2, Reporter arg3)
				throws IOException {
			// TODO Auto-generated method stub
			
		}
		
	}
	
	public static class ParrallelizedSRTReducer extends MapReduceBase
     implements Reducer<IntWritable, ConfigGraph, Text, Text> {

		public void reduce(IntWritable key, Iterator<ConfigGraph> values,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {

			boolean first = true;
			StringBuilder toReturn = new StringBuilder();
			while (values.hasNext()){
				if (!first)
					toReturn.append(", ");
				first=false;
				toReturn.append(values.next().toString());
			}

			output.collect(key, new Text(toReturn.toString()));
		}
 }

	/**
	 * @param args
	 */
	public static void main(String[] args) {
	    JobClient client = new JobClient();
	    JobConf conf = new JobConf(ParrallelizedSRT.class);

	    conf.setJobName("LineIndexer");

	    conf.setOutputKeyClass(Text.class);
	    conf.setOutputValueClass(Text.class);

	    FileInputFormat.addInputPath(conf, new Path("input"));
	    FileOutputFormat.setOutputPath(conf, new Path("output"));

	    conf.setMapperClass(ParrallelizedSRTMapper.class);
	    conf.setReducerClass(ParrallelizedSRTReducer.class);

	    client.setConf(conf);

	    try {
	      JobClient.runJob(conf);
	    } catch (Exception e) {
	      e.printStackTrace();
	    }

	}

}
